package com.zhangxun.service.impl;


import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.rabbitmq.client.Channel;
import com.zhangxun.constant.QueueConstant;
import com.zhangxun.doc.ProdEs;
import com.zhangxun.domain.Prod;
import com.zhangxun.domain.ProdChange;
import com.zhangxun.domain.ProdComm;
import com.zhangxun.domain.ProdTagReference;
import com.zhangxun.esinterface.EsRepository;
import com.zhangxun.service.*;
import com.zhangxun.utils.ThreadPoolUtil;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

@Service
public class EsSImportService {

    @Value("${esimport.size}")
    private Integer importSize;



    @Autowired
    private ProdService prodService;



    @Autowired
    private EsRepository esRepository;

    private Date startDate;


    /*
     查询出全部未上架的商品有商品，并给商品填充需要放入到es当中的属性值
     这样有缺陷，商品太多。list太大容易oom。
    public void selectAllProd() {
        //查询出所有未上架的商品
        List<Prod> prodList = prodService.list(new LambdaQueryWrapper<Prod>().eq(Prod::getStatus, 0));
        if (CollectionUtils.isEmpty(prodList)) {
            return;
        }
        //给每个商品设置标签
        setTagIdByProdId(prodList);
        //给每个商品设置好评数和好评率
        setPraiseNumAndPositiveRat(prodList);
        int size = prodList.size();
        int i = 0;
        while (true) {
            if ((i + importSize) >= size) {
                elasticsearchRestTemplate.save(prodList.subList(i,size));
                break;
            }
            elasticsearchRestTemplate.save(prodList.subList(i, i + importSize));
            i += importSize;
        }
    }
     */



    /**
     * 异步操作分批次查询数据导入到es当中,全量
     */
    @PostConstruct
    public void importProdToES(){
        //查询出未上架商品总数，为分页查询做准备
        int count = prodService.count(new LambdaQueryWrapper<Prod>()
                .eq(Prod::getStatus,0));
        //判断是否有未上架的商品，如果没有就直接返回
        if(count <= 0){
            return;
        }
        //计算循环的次数
        int page = count%importSize == 0? count/importSize:(count/importSize +1);
        CountDownLatch countDownLatch = new CountDownLatch(page);
        //循环分批次导入商品
        for(int i = 1;i<= page ;i++){
            int current = i;
            //异步查询商品，并导入到es的方法
            ThreadPoolUtil.threadPool().execute(() -> {
                    prodService.importToES(current,importSize,null,null);
                    countDownLatch.countDown();
                    });
        }

        //当前程序等待异步导入到es的操作执行完再执行
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        //设置增量导入查询数据库新增商品的开始时间
        startDate = new Date();

    }

    /**
     *  增量导入方法，定时任务
     */
    @Scheduled(fixedDelay= 1000*60*15,initialDelay=1000*60)
    public void addImport(){

        Date endDate = new Date(); //设置当前时间为查询结束时间。
        //查询出未上架商品总数，为分页查询做准备
        int count = prodService.count(new LambdaQueryWrapper<Prod>()
                .eq(Prod::getStatus,0)
                .between(Prod::getCreateTime,startDate,endDate));
        //判断是否有未上架的商品，如果没有就直接返回
        if(count <= 0){
            return;
        }
        //计算循环的次数
        int page = count%importSize == 0? count/importSize:(count/importSize +1);
        CountDownLatch countDownLatch = new CountDownLatch(page);
        //循环分批次导入商品
        for(int i = 1;i<= page ;i++){
            int current = i;

            //异步查询商品，并导入到es
            ThreadPoolUtil.threadPool().execute(() -> {
                prodService.importToES(current,importSize,startDate,endDate);
                countDownLatch.countDown();
            });
        }
        //当前程序等待异步导入到es的操作执行完再执行
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        //重新设置开始时间
        startDate = endDate;
    }

    /**
     * 使用消息队列扣减es商品库存。
     * @param message
     * @param channel
     *
     * ProdChange 封装了用户购买或者退款的商品的prodId和数量count。
     * count为正数则为退款，添加库存。
     * count为负数则为购买，扣减库存。
     * 用户可能一次性购买多个商品，所以传输一个集合封装购买的所有商品
     *
     */
    @RabbitListener(queues = QueueConstant.ES_PROD_OPERATION_QUEUE)
    public void operationProd(Message message, Channel channel){
        //获取封装数据的二进制流，转换成字符串，再转换成集合。
        String str = new String(message.getBody());
        List<ProdChange> prodChangeList = JSON.parseArray(str, ProdChange.class);
        //转换成商品id的集合
        List<Long> prodIdList = prodChangeList.stream()
                .map(ProdChange::getProdId).collect(Collectors.toList());
        //通过商品id查询es查询到所有对应id的商品。
        List<ProdEs> prodEsList = (List<ProdEs>) esRepository.findAllById(prodIdList);

        //将购买的商品list对象转换成map对象方便查询
        Map<Long, ProdChange> map = prodChangeList.stream()
                .collect(Collectors.toMap(ProdChange::getProdId, prodChange -> prodChange));

        //遍历商品es对象，进行判断和扣减库存操作。
        prodEsList.stream().forEach(prodEs->{

            //获取当前商品对应的购买数量的对象
            ProdChange prodChange = map.get(prodEs.getProdId());

            //获取库存量加上购买或者退货量的结果，购买count为负数。+可以满足购买和退货。
            long num = prodEs.getTotalStocks()+prodChange.getCount();

            //判断库存是否充足，是就扣减es库存。
            if( num>=0 ){
                prodEs.setTotalStocks(num);
                esRepository.save(prodEs);
            }

        });

        //签收消息
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}
